package com.yanggu.flink.datastream_api.test


import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.{Collector, OutputTag}

/**
 * 测试用的ProcessFunction, 使用于DataStream上的ProcessFunction
 */
class PassThroughProcessFunction extends ProcessFunction[Integer, Integer] {

    @throws[Exception]
    override def processElement(value: Integer, ctx: ProcessFunction[Integer, Integer]#Context, out: Collector[Integer]): Unit = {
      //向下游输出
      out.collect(value)
      //使用侧输出流
      ctx.output(new OutputTag[Integer]("test-output-tag", Types.INT), value)
    }

}